package com.gdlt.mq2db;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.hadoop.conf.Configuration;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/**
 * Created by CM on 2017/3/14.
 */
public class PGWriter implements Offsetable, Writable {
	Connection connection;
	String topic;
	int partitionId;

	private String recValue;
	private String ip;
	private String index;
	private BigDecimal value;// 对应数据库的decimal类型，若用double精度显示不全
	private int ts;
	private static AtomicBoolean flag = new AtomicBoolean(true);
	//boolean closeFlag = false; //标志位

	private Connection getConn() { // 得到数据库的连接

		Configuration conf = Kafka2pgConfig.getConf();
		String url = conf.get(Kafka2pgConfig.PG_URL_CONF_NAME);
		String user = conf.get(Kafka2pgConfig.PG_USER_CONF_NAME);
		String password = conf.get(Kafka2pgConfig.PG_PASSWORD_CONF_NAME);
		String driver = conf.get(Kafka2pgConfig.PG_DRIVER_CONF_NAME);
		Connection connection = null;// 数据库连接
		try {
			Class.forName(driver);// 加载驱动程序，你的jar包要用上啊
			connection = DriverManager.getConnection(url, user, password);
			System.out.println("成功连接pg数据库:" + connection);
		} catch (ClassNotFoundException e) {
			// LOG.error("未加载数据库驱动程序" + e);
		} catch (SQLException e) {
			// LOG.error("连接数据库异常" + e);
		}
		return connection;
	}

	 public PGWriter(int partitionId){
		 this.partitionId = partitionId;
	
	 }

	/* (non-Javadoc)
	 * @see com.gdlt.mq2db.Offsetable#getLastoffset()
	 */	
	public long getLastoffset() { 
		long lastOffset = -1;
		connection = getConn();
		PreparedStatement pstmt = null;
		String sql = "select max(recset) from golia_offset";

		try {
			pstmt = connection.prepareStatement(sql);
			ResultSet rs = pstmt.executeQuery();
			if (rs.next()&&rs.getObject(1)!=null) {//由于统计结果即使没记录也会返回null，所以加多个判断条件
			String	strOffset = rs.getString(1);
			lastOffset = Long.parseLong(strOffset);
			}
			System.out.println("读取数据库中offset:"+lastOffset);

		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return lastOffset;
	}

	
	/* (non-Javadoc)
	 * @see com.gdlt.mq2db.Writable#write(java.util.List, long)
	 */
	
	public boolean write(List<ConsumerRecord<String, String>> data, long offset) {
		boolean writeState = false;
		connection = getConn();// 获得数据库连接
		PreparedStatement pstmt = null; // 数据库操作，插入数据到golia_data表中
		PreparedStatement pstmtSet = null; // 数据库操作，插入数据到golia_offset表中
//		String sql = "insert into golia_data(ip,index,value,ts) select ?,?,?,?"
//				+ "where not exists(select ip,index,ts from golia_data where " + "ip = ? and index = ? and ts = ?)";
		String sql = "insert into golia_data(ip,index,value,ts) values(?,?,?,?)"
				+ "on conflict (ip,index,ts) do nothing ";
		
		String sqlSet = "insert into golia_offset(recset,partition) values(?,?)"
				+ "on conflict(recset,partition) do nothing";

		try {
			connection.setAutoCommit(false);
			pstmt = connection.prepareStatement(sql);
			pstmtSet = connection.prepareStatement(sqlSet);

			for (int i = 0; i < data.size(); i++) { // 先对消息队列中的value值根据空格划分
				recValue = data.get(i).value();
				String splitBlank[] = recValue.split(" ");
				String splitComma[] = splitBlank[0].split("\\.");// 根据点号分隔

				ip = splitComma[1].replace('_', '.');
				index = splitBlank[0].substring(splitComma[0].length() + splitComma[1].length() + 2);
				value = new BigDecimal(splitBlank[1]);// 传入String
				ts = Integer.parseInt(splitBlank[2]);

				pstmt.setString(1, ip);
				pstmt.setString(2, index);
				pstmt.setBigDecimal(3, value);
				pstmt.setInt(4, ts);
				pstmt.addBatch();

			}
			int[] counts = pstmt.executeBatch();			 
			String offsetStr = String.valueOf(offset);
			//offset = data.get(data.size() - 1).offset();
			pstmtSet.setString(1, offsetStr);
			pstmtSet.setInt(2, partitionId);
			pstmtSet.addBatch();
			pstmtSet.executeUpdate();
			
			connection.commit();// 一起提交
			
			int dupli = 0, fails = 0;
			for (int i = 0; i < counts.length; i++) {
				if (counts[i] == 0) {
					dupli++;
				}

				if (counts[i] == Statement.EXECUTE_FAILED) {
					fails++;
				}
			}
			int inPg = counts.length - dupli - fails;
			 System.out.println("这批数据共" + counts.length + "条，插入数据库有" +
			 inPg + "条，其中有" + dupli + "条记录键值重复，" + fails + "条记录执行失败。");
			 writeState= true;
			pstmt.close(); // 操作关闭
			pstmtSet.close(); // 操作关闭
			
		} catch (SQLException e) {
			try {
				
				connection.rollback();// 回滚操作
				
				// LOG.info("回滚成功");
			} catch (Exception e2) {
				// TODO: handle exception
				// LOG.error("回滚异常" + e2);
			}
			// TODO Auto-generated catch block
			// LOG.error("插入数据库异常:" + e);
		}
		return writeState;

	}

	
	/* (non-Javadoc)
	 * @see com.gdlt.mq2db.Writable#isClosed()
	 */
	public boolean isClosed(){		
		return flag.get();
	}
	
	
	/* (non-Javadoc)
	 * @see com.gdlt.mq2db.Writable#close()
	 */
	public void close() {
		try {	
			//就是说这个从true到false的过程才算被关闭	
			if(flag.compareAndSet(true, false)){
			connection.close();
			}
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
